package com.kele.test.rabbitmq.service;

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

@Component
@RabbitListener(queues = {"Queue_reliable"},concurrency = "2")
public class MQService_High5_Reliable {

    List<String> revMessageIdList = new ArrayList<>();
    /**
     * 设置为手动ACK模式
     * @param msg
     * @param message
     * @param channel
     * @throws IOException
     */
    @RabbitHandler()
    public void testRevMsg(String msg, Message message, Channel channel) throws IOException {

        //自动发送ACK ，MQ接收到ACK后，将删除消息
        /*修改为手动 ACK
        listener:
          simple:
            acknowledge-mode: manual
         *///int i = 10/0;
        String messageId = message.getMessageProperties().getMessageId();

        System.out.println("Thread:"+Thread.currentThread().getId()
                +" Queue_reliable ++++++" + msg + " messageId=" + messageId);
        //从数据库 查询messageId 的状态 ，
        //if(status==1) 跳过执行业务逻辑
        //else 执行业务逻辑
        //保证该消息不会重复消费
        synchronized (revMessageIdList){
            if(revMessageIdList.contains(messageId)){
                System.out.println("该消息重复接收...");
            }else {
                System.out.println("执行完业务逻辑");
                try {
                    Thread.sleep(10);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

                revMessageIdList.add(messageId);// 保存到数据库 指示 messageId  该条消息 status = 1//已经处理
            }
        }


        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        channel.basicAck(deliveryTag,false);//ACK
    }
}
